package org.apache.ocean.main.search;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

import org.apache.ocean.main.Snapshot;
import org.apache.ocean.main.Index.IndexSnapshot;
import org.apache.ocean.main.fields.DoubleFieldType;
import org.apache.ocean.main.fields.FieldType;
import org.apache.ocean.main.fields.LongFieldType;
import org.apache.ocean.main.search.cache.FieldCacheKey.FieldIndex;
import org.apache.ocean.main.util.Timeout;

public class SnapshotSearch extends Search {
	private Snapshot snapshot;
	private ExecutorService executorService;
	private long startTime;

	public SnapshotSearch(SearchQuery searchQuery, Snapshot snapshot, ExecutorService executorService) {
		super(searchQuery);
		this.snapshot = snapshot;
		this.executorService = executorService;
	}

	// TODO: try using multireader for non sort queries
	public Result<ScoreDoc> call() throws Exception {
		startTime = Timeout.TIMER_THREAD.getTime();
		Collection<IndexSnapshot> indexSnapshots = snapshot.getIndexSnapshots();
		List<Callable<Result<ScoreDoc>>> indexSearches = new ArrayList<Callable<Result<ScoreDoc>>>(indexSnapshots.size());
		for (IndexSnapshot indexSnapshot : indexSnapshots) {
			IndexSearch indexSearch = new IndexSearch(searchQuery, indexSnapshot);
			indexSearches.add(indexSearch);
		}
		List<Future<Result<ScoreDoc>>> futures = executorService.invokeAll(indexSearches);
		List<Result<ScoreDoc>> results = new ArrayList<Result<ScoreDoc>>(indexSnapshots.size());
		long aggregateTotalLong = 0;
		double aggregateTotalDouble = 0;
		Aggregate aggregate = searchQuery.getAggregate();
		for (Future<Result<ScoreDoc>> future : futures) {
			Result<ScoreDoc> result = future.get();
			if (aggregate != null && (aggregate.isSum() || aggregate.isAvg())) {
				Object value = result.getAggregateResult().getValue();
				if (value instanceof Double) {
					aggregateTotalDouble += ((Double) value).doubleValue();
				} else if (value instanceof Long) {
					aggregateTotalLong += ((Long) value).longValue();
				}
			}
			results.add(result);
		}
		if (aggregate != null) {
			Result<ScoreDoc> mergedResult = mergeScore(results);
			FieldType fieldType = snapshot.getIndexSchemaCategory().getField(aggregate.getField()).getFieldType();
			Object value = null;
			// TODO: move aggregation to own class
			if (fieldType instanceof DoubleFieldType) {
				if (aggregate.isAvg()) {
					value = aggregateTotalDouble / mergedResult.getNumFound();
				} else {
					value = aggregateTotalDouble;
				}
			} else if (fieldType instanceof LongFieldType) {
				if (aggregate.isAvg()) {
					value = aggregateTotalLong / mergedResult.getNumFound();
				} else {
					value = aggregateTotalLong;
				}
			}
			AggregateResult aggregateResult = new AggregateResult(value, aggregate);
			mergedResult.setAggregateResult(aggregateResult);
			return mergedResult;
		} else if (searchQuery.hasSort()) {
			return mergeSort(searchQuery.getSort(), results);
		} else {
			return mergeSort(Sort.SCORE_SORT, results);
		}
	}
	
	// TODO: use only merge sort
	private Result<ScoreDoc> mergeSort(Sort sort, List<Result<ScoreDoc>> results) throws Exception {
		int numDocs = searchQuery.getNumDocs();
		Crowds crowds = searchQuery.getCrowds();
		Map<CrowdKey, CrowdValueResult> crowdMap = new HashMap<CrowdKey, CrowdValueResult>(100);
		FieldSortedQueue fieldSortedQueue = new FieldSortedQueue(sort, numDocs);
		int numFound = 0;
		for (Result<ScoreDoc> result : results) {
			numFound += result.getNumFound();
			ScoreDoc[] scoreDocs = result.getAllDocs();
			for (ScoreDoc scoreDoc : scoreDocs) {
				if (crowds != null) {
					
					for (int x = 0; x < crowds.size(); x++) {
						Crowd crowd = crowds.get(x);
						FieldIndex idFieldIndex = scoreDoc.indexSnapshot.getFieldIndex(crowd.getField());
						Long id = (Long) idFieldIndex.getValue(scoreDoc.doc);
						Comparable crowdValue = scoreDoc.crowdValues[x];
						int maxCount = crowd.getMaxCount();
						//CrowdValueKey crowdValueKey = new CrowdValueKey(crowd.getField(), crowdValue);
						CrowdKey crowdKey = new CrowdKey(crowd.getField(), crowdValue);
						CrowdValueResult crowdValueResult = crowdMap.get(crowdValueKey);
						if (crowdValueResult == null) {
							crowdValueResult = new CrowdValueResult(id, crowd, crowdValue);
							crowdMap.put(crowdValueKey, crowdValueResult);
						}
						if (crowdValueResult.getCount() < maxCount) {
							fieldSortedQueue.insert(scoreDoc);
						}
						crowdValueResult.incrementCount();
					}
				} else {
					fieldSortedQueue.insert(scoreDoc);
				}
			}
		}
		ScoreDoc[] mergedScoreDocs = toArray(fieldSortedQueue);
		long elapsed = Timeout.TIMER_THREAD.getTime() - startTime;
		Result<ScoreDoc> mergedResult = new Result<ScoreDoc>(searchQuery.getStart(), mergedScoreDocs, numFound, elapsed);
		return mergedResult;
	}
	/**
	 * private Result<ScoreDoc> mergeScore(List<Result<ScoreDoc>> results)
	 * throws Exception {
	 * 
	 * int numDocs = searchQuery.getNumDocs(); final ScorePriorityQueue
	 * scorePriorityQueue = new ScorePriorityQueue(numDocs); int numFound = 0; for
	 * (Result<ScoreDoc> result : results) { numFound += result.getNumFound();
	 * ScoreDoc[] scoreDocs = result.getAllDocs(); for (ScoreDoc scoreDoc :
	 * scoreDocs) { if (crowds != null) { for (int x = 0; x < crowds.size(); x++) {
	 * Crowd crowd = crowds.get(x); FieldIndex idFieldIndex =
	 * scoreDoc.indexSnapshot.getFieldIndex(crowd.getField()); Long id = (Long)
	 * idFieldIndex.getValue(scoreDoc.doc); Comparable crowdValue =
	 * scoreDoc.crowdValues[x]; int maxCount = crowd.getMaxCount(); CrowdValueKey
	 * crowdValueKey = new CrowdValueKey(crowd.getField(), crowdValue);
	 * CrowdValueResult crowdValueResult = crowdMap.get(crowdValueKey); if
	 * (crowdValueResult == null) { crowdValueResult = new CrowdValueResult(id,
	 * crowd, crowdValue); crowdMap.put(crowdValueKey, crowdValueResult); } if
	 * (crowdValueResult.getCount() < maxCount) {
	 * scorePriorityQueue.insert(scoreDoc); } crowdValueResult.incrementCount(); } }
	 * else { scorePriorityQueue.insert(scoreDoc); } } } ScoreDoc[]
	 * mergedScoreDocs = toArray(scorePriorityQueue); long elapsed =
	 * Timeout.TIMER_THREAD.getTime() - startTime; Result<ScoreDoc> mergedResult =
	 * new Result(searchQuery.getStart(), mergedScoreDocs, numFound, elapsed);
	 * return mergedResult; }
	 */
}
